package com.market_analysis

import java.sql.Timestamp
import java.util.UUID

import com.market_analysis.bean.{MarketUserBehavior, MarketViewCount}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.util.Random

/**
  * @Description: TODO QQ1667847363
  * @author: xiao kun tai
  * @date:2021/12/4 9:26
  *                按APP渠道和行为统计数据
  */
object AppMarketByChannel {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) //定义事件时间语义

    val dataStream: DataStream[MarketUserBehavior] = env.addSource(new SimulatedSource)
      .assignAscendingTimestamps(_.timestamp)

    //开窗统计输出
    val resultStream: DataStream[MarketViewCount] = dataStream.filter(_.behavior != "uninstall")
      .keyBy(data => (data.behavior, data.channel))
      .timeWindow(Time.days(1), Time.seconds(5))
      .process(new MarketCountByChannel)

    resultStream.print("app")

    env.execute("app market by channel job")


  }

  //自定义Source数据源
  class SimulatedSource extends RichSourceFunction[MarketUserBehavior] {
    //是否运行的标志位
    var running = true
    //定义用户行为和渠道的集合
    val behaviorSet: Seq[String] = Seq("view", "download", "install", "uninstall")
    val channelSet: Seq[String] = Seq("appstore", "weibo", "wechat", "bilibili", "tieba")
    val rand: Random = Random

    override def run(sourceContext: SourceFunction.SourceContext[MarketUserBehavior]): Unit = {
      //定义一个生成数据最大的数量
      val maxCounts = Long.MaxValue
      var count = 0L
      while (running && count < maxCounts) {
        val id = UUID.randomUUID().toString
        val behavior = behaviorSet(rand.nextInt(behaviorSet.size))
        val channel = channelSet(rand.nextInt(channelSet.size))
        val ts = System.currentTimeMillis()
        sourceContext.collect(MarketUserBehavior(id, behavior, channel, ts))
        count += 1
        Thread.sleep(50L)
      }
    }

    override def cancel(): Unit = running = false
  }

  //自定义ProcessWindowFunction
  class MarketCountByChannel extends ProcessWindowFunction[MarketUserBehavior, MarketViewCount, (String, String), TimeWindow] {
    override def process(key: (String, String), context: Context, elements: Iterable[MarketUserBehavior], out: Collector[MarketViewCount]): Unit = {
      val start = new Timestamp(context.window.getStart).toString
      val end = new Timestamp(context.window.getEnd).toString
      val channel = key._1
      val behavior = key._2
      val count = elements.size
      out.collect(MarketViewCount(start, end, channel, behavior, count))

    }
  }

}
